package com.smartboot.plus.features.metric;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.smartboot.plus.condition.ConditionGroup;
import com.smartboot.plus.condition.ConditionQ;
import com.smartboot.plus.dto.MetricDO;
import com.smartboot.plus.dto.MetricItemDTO;
import com.smartboot.plus.enums.BrokerStatueEnum;
import com.smartboot.plus.enums.MqttMetricEnum;
import com.smartboot.plus.features.NodeUtil;
import com.smartboot.plus.ignite.cache.MqttMetricCache;
import com.smartboot.plus.plugins.PluginConfig;
import com.smartboot.plus.service.MqttMetricService;
import com.smartboot.plus.service.MqttNodeService;
import com.smartboot.plus.utils.DateUtil;
import com.smartboot.plus.utils.Func;
import com.smartboot.plus.utils.SystemInfoUtil;
import org.smartboot.socket.StateMachineEnum;
import org.smartboot.socket.extension.plugins.AbstractPlugin;
import org.smartboot.socket.transport.AioSession;
import tech.smartboot.feat.cloud.annotation.Autowired;
import tech.smartboot.feat.cloud.annotation.Bean;
import tech.smartboot.feat.cloud.annotation.PostConstruct;
import tech.smartboot.feat.core.common.logging.Logger;
import tech.smartboot.feat.core.common.logging.LoggerFactory;
import tech.smartboot.mqtt.common.message.MqttConnAckMessage;
import tech.smartboot.mqtt.common.message.MqttConnectMessage;
import tech.smartboot.mqtt.common.message.MqttMessage;
import tech.smartboot.mqtt.common.message.MqttPublishMessage;
import tech.smartboot.mqtt.plugin.spec.BrokerContext;
import tech.smartboot.mqtt.plugin.spec.Message;
import tech.smartboot.mqtt.plugin.spec.MqttSession;
import tech.smartboot.mqtt.plugin.spec.bus.EventObject;
import tech.smartboot.mqtt.plugin.spec.bus.*;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.Timestamp;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
import java.util.stream.Collectors;

@Bean
public class MqttMetricFeatures {

    private static final Logger LOGGER = LoggerFactory.getLogger(MqttMetricFeatures.class);
    private final Map<MqttMetricEnum, List<MetricDO>> metricMap = new HashMap();

    private static Map<MqttMetricEnum, MetricItemDTO> metrics = new ConcurrentHashMap();
    private static final long START_TIME = System.currentTimeMillis();


    @Autowired
    private BrokerContext brokerContext;

    @Autowired
    private PluginConfig pluginConfig;


    @Autowired
    private MqttMetricService mqttMetricService;


    @Autowired
    private MqttNodeService mqttNodeService;

    @PostConstruct
    public void init() {
        try {
            initMetric();
            metricsTimer();
            collectClusterMetrics();
        } catch (Exception e) {
            LOGGER.error("init metric error", e);
        }
        brokerContext.Options().addPlugin(new AbstractPlugin<MqttMessage>() {
            @Override
            public void afterRead(AioSession session, int readSize) {
                if (readSize > 0) {
                    // 已接收字节数
                    MetricItemDTO metric = metrics.get(MqttMetricEnum.BYTES_RECEIVED);
                    if (metric != null) {
                        metric.getMetric().add(readSize);
                    }
                }
            }

            @Override
            public void afterWrite(AioSession session, int writeSize) {
                if (writeSize > 0) {
                    // 已发送字节数
                    MetricItemDTO metric = metrics.get(MqttMetricEnum.BYTES_SENT);
                    if (metric != null) {
                        metric.getMetric().add(writeSize);
                    }
                }
            }

            @Override
            public void stateEvent(StateMachineEnum stateMachineEnum, AioSession session, Throwable throwable) {
                if (stateMachineEnum == StateMachineEnum.NEW_SESSION) {
                    MetricItemDTO metric = metrics.get(MqttMetricEnum.CLIENT_ONLINE);
                    if (metric != null) {
                        metric.getMetric().increment();
                    }
                } else if (stateMachineEnum == StateMachineEnum.SESSION_CLOSED) {
                    MetricItemDTO metric = metrics.get(MqttMetricEnum.CLIENT_ONLINE);
                    if (metric != null) {
                        metric.getMetric().decrement();
                    }
                }
            }
        });
    }

    public void metricsTimer() {


        //定时
        brokerContext.getTimer().scheduleWithFixedDelay(() -> {
            // 获取实际发送的消息数量
            MetricItemDTO metricItemDTO = metrics.get(MqttMetricEnum.PACKETS_PUBLISH_SENT);
            // 计算已发送的消息数量 (当前值 - 最新值)
            long sent = metricItemDTO.getValue() - metricItemDTO.getLatestValue();

            // 获取预期发送的消息数量 订阅者数量
            MetricItemDTO metricItemDTO1 = metrics.get(MqttMetricEnum.PACKETS_EXPECT_PUBLISH_SENT);
            // 计算预期发送的消息数量 (当前值 - 最新值)
            long expect = metricItemDTO1.getValue() - metricItemDTO1.getLatestValue();

            // 如果预期发送数量大于 0 且实际发送数量小于预期数量
            if (expect > 0L && sent < expect) {
                // 计算发送速率，并乘以 1000
                //sent 是实际发送的数量。
                //expect 是预期的数量。
                //结果将是 sent 占 expect 的百分比。
                long rate = (sent * 1000) / expect; // 计算百分比
                // 更新发送速率
                metrics.get(MqttMetricEnum.PACKETS_PUBLISH_RATE).setValue(rate);
            } else {
                if (expect == 0 && sent == 0 && expect == 0) {
                    // 如果发送正常，直接将速率设置为 1000
                    metrics.get(MqttMetricEnum.PACKETS_PUBLISH_RATE).setValue(0L);
                } else {
                    metrics.get(MqttMetricEnum.PACKETS_PUBLISH_RATE).setValue(1000L);
                }
            }

            Iterator iterator = metrics.entrySet().iterator();

            while (iterator.hasNext()) {
                Map.Entry<MqttMetricEnum, MetricItemDTO> entry = (Map.Entry) iterator.next();
                MqttMetricEnum metric = entry.getKey();
                MetricItemDTO value = entry.getValue();
                MetricDO metricDO = new MetricDO();

                metricDO.setNodeName(brokerContext.Options().getNodeId());
                metricDO.setObjectType("node");
                metricDO.setObjectId(brokerContext.Options().getNodeId());
                metricDO.setCode(metric.getCode());
                long currentValue = value.getValue();
                if (metric.isPeriodRest()) {
                    metricDO.setValue(currentValue - value.getLatestValue());
                } else {
                    metricDO.setValue(currentValue);
                }

                value.setLatestValue(currentValue);
                List<MetricDO> list = metricMap.computeIfAbsent(entry.getKey(), (mqttMetricEnum) -> {
                    return new LinkedList();
                });
                if (list.size() >= 16) {
                    list.remove(0);

                }
                Date date = new Date(System.currentTimeMillis() / 5000L * 5000L);
                metricDO.setCreateTime(DateUtil.format(date, DateUtil.PATTERN_DATETIME));
                list.add(metricDO);

            }

        }, 5L, TimeUnit.SECONDS);


    }


    private void initMetric() {
        // 初始化 metrics
        MqttMetricEnum[] var2 = MqttMetricEnum.values();
        for (MqttMetricEnum metricEnum : var2) {
            metrics.put(metricEnum, new MetricItemDTO(metricEnum));
        }
        EventBus eventBus = brokerContext.getEventBus();
        eventBus.subscribe(EventType.CONNECT, (eventType, object) -> {
            //客户端连接次数
            incrementMetric(MqttMetricEnum.CLIENT_CONNECT);
        });


        eventBus.subscribe(EventType.DISCONNECT, (eventType, session) -> {
            //客户端断开连接次数
            incrementMetric(MqttMetricEnum.CLIENT_DISCONNECT);
        });


        eventBus.subscribe(EventType.SUBSCRIBE_ACCEPT, (eventType, object) -> {
            //订阅次数
            incrementMetric(MqttMetricEnum.CLIENT_SUBSCRIBE);
        });
        eventBus.subscribe(EventType.UNSUBSCRIBE_ACCEPT, (eventType, object) -> {
            //取消订阅次数
            incrementMetric(MqttMetricEnum.CLIENT_UNSUBSCRIBE);
        });
        eventBus.subscribe(EventType.SUBSCRIBE_TOPIC, (eventType, object) -> {
            //订阅关系数
            incrementMetric(MqttMetricEnum.SUBSCRIBE_RELATION);
        });
        eventBus.subscribe(EventType.UNSUBSCRIBE_TOPIC, (eventType, object) -> {
            //取消订阅
            decrementMetric(MqttMetricEnum.SUBSCRIBE_RELATION);

        });
        eventBus.subscribe(EventType.TOPIC_CREATE, (eventType, object) -> {
            // 获取 TOPIC_COUNT 的计数器，确保不为 null
            incrementMetric(MqttMetricEnum.TOPIC_COUNT);
        });

        //接收到客户端发送的任何消息
        eventBus.subscribe(EventType.RECEIVE_MESSAGE, new EventBusConsumer<EventObject<MqttMessage>>() {
            //接收的报文数量
            LongAdder packetsReceived = metrics.get(MqttMetricEnum.PACKETS_RECEIVED).getMetric();
            //接收的 CONNECT 报文数量
            LongAdder connectReceived = metrics.get(MqttMetricEnum.PACKETS_CONNECT_RECEIVED).getMetric();

            public void consumer(EventType<EventObject<MqttMessage>> eventType, EventObject<MqttMessage> object) {
                this.packetsReceived.increment();
                if (object.getObject() instanceof MqttConnectMessage) {
                    this.connectReceived.increment();
                }
            }
        });

        //往客户端发送的任何消息
        eventBus.subscribe(EventType.WRITE_MESSAGE, new EventBusConsumer<EventObject<MqttMessage>>() {


            LongAdder packetsSent = metrics.get(MqttMetricEnum.PACKETS_SENT).getMetric();
            LongAdder connAckSent = metrics.get(MqttMetricEnum.PACKETS_CONNACK_SENT).getMetric();
            LongAdder publishSent = metrics.get(MqttMetricEnum.PACKETS_PUBLISH_SENT).getMetric();
            LongAdder qos0Sent = metrics.get(MqttMetricEnum.MESSAGE_QOS0_SENT).getMetric();
            LongAdder qos1Sent = metrics.get(MqttMetricEnum.MESSAGE_QOS1_SENT).getMetric();
            LongAdder qos2Sent = metrics.get(MqttMetricEnum.MESSAGE_QOS2_SENT).getMetric();


            @Override
            public void consumer(EventType<EventObject<MqttMessage>> eventType, EventObject<MqttMessage> object) {
                this.packetsSent.increment();

                if (object.getObject() instanceof MqttConnAckMessage) {
                    this.connAckSent.increment();
                } else if (object.getObject() instanceof MqttPublishMessage) {
                    this.publishSent.increment();
                    switch ((object.getObject()).getFixedHeader().getQosLevel()) {
                        case AT_MOST_ONCE:
                            this.qos0Sent.increment();
                            break;
                        case AT_LEAST_ONCE:
                            this.qos1Sent.increment();
                            break;
                        case EXACTLY_ONCE:
                            this.qos2Sent.increment();
                            break;
                        default:
                            throw new IllegalStateException();
                    }
                }

            }
        });


        //生产
        brokerContext.getMessageBus().consumer(new MessageBusConsumer() {
            // 获取计数器
            LongAdder publishReceived = metrics.get(MqttMetricEnum.PACKETS_PUBLISH_RECEIVED).getMetric();
            LongAdder expectPublishSent = metrics.get(MqttMetricEnum.PACKETS_EXPECT_PUBLISH_SENT).getMetric();
            LongAdder qos0Received = metrics.get(MqttMetricEnum.MESSAGE_QOS0_RECEIVED).getMetric();
            LongAdder qos1Received = metrics.get(MqttMetricEnum.MESSAGE_QOS1_RECEIVED).getMetric();
            LongAdder qos2Received = metrics.get(MqttMetricEnum.MESSAGE_QOS2_RECEIVED).getMetric();

            public void consume(MqttSession session, Message publishMessage) {
                this.publishReceived.increment();

                switch (publishMessage.getQos()) {
                    case AT_MOST_ONCE:
                        this.qos0Received.increment();
                        break;
                    case AT_LEAST_ONCE:
                        this.qos1Received.increment();
                        break;
                    case EXACTLY_ONCE:
                        this.qos2Received.increment();
                        break;
                    default:
                        throw new IllegalStateException();
                }
            }
        });
    }


    // 公共的计数器增加方法
    private void incrementMetric(MqttMetricEnum metricEnum) {
        MetricItemDTO metricItem = metrics.get(metricEnum);
        if (metricItem != null) {
            metricItem.getMetric().increment();
        }
    }

    // 公共的计数器减少方法
    private void decrementMetric(MqttMetricEnum metricEnum) {
        MetricItemDTO metricItem = metrics.get(metricEnum);
        if (metricItem != null) {
            if (metricItem.getMetric().intValue() > 0) {
                metricItem.getMetric().decrement();
            }
        }
    }


    private void collectClusterMetrics() {

        List<String> metrics = Func.toStrList("packets_expect_publish_sent,packets_received,packets_publish_rate,packets_publish_received,packets_publish_sent,topic_count,client_online,subscribe_relation,bytes_received,bytes_sent,packets_sent");
        brokerContext.getTimer().scheduleWithFixedDelay(() -> {
            try {
                // 根据 recordTypeEnum 判断数据源
                List<MetricDO> publishReceiveCount = new ArrayList<>();
                if (System.currentTimeMillis() - START_TIME > 10000L) {
                    metrics.forEach((code) -> {
                        List<MetricDO> metricData = metricMap.get(MqttMetricEnum.getByCode(code));
                        if (metricData != null) {
                            publishReceiveCount.addAll(metricData);
                        }
                    });
                }
                // 创建返回的 JSON 对象
                JSONObject jsonObject = new JSONObject();
                // 按照 code 分组
                Map<String, List<MetricDO>> collect = publishReceiveCount.stream()
                        // 按照 code 分组
                        .collect(Collectors.groupingBy(MetricDO::getCode));

                collect.forEach((metric, list) -> {
                    // 按照 createTime 分组
                    Map<String, Map<String, Object>> group = new HashMap<>();
                    list.stream()
                            .collect(Collectors.groupingBy(MetricDO::getCreateTime)) // 按照 createTime 分组
                            .forEach((date, sublist) -> {
                                // 每个时间段对应的数据
                                Map<String, Object> nodesMetric = new HashMap<>();
                                nodesMetric.put("date", date);
                                // 将每个节点的值放到结果中
                                sublist.forEach(metricDO -> {
                                    nodesMetric.put("value", metricDO.getValue());
                                });
                                // 将数据按时间插入 group 中
                                group.put(date, nodesMetric);
                            });
                    // 按时间排序
                    List<Map<String, Object>> sortedList = new ArrayList<>();
                    group.keySet().stream()
                            .sorted() // 按时间排序
                            .forEach(date -> sortedList.add(group.get(date)));
                    // 将结果放入 jsonObject 中
                    jsonObject.put(metric, sortedList);
                });


                Map<String, ClientMetrics> result = new HashMap<>();
                // 遍历每个 JSONArray 并处理
                for (String key : jsonObject.keySet()) {
                    JSONArray metricArray = jsonObject.getJSONArray(key);
                    // Create an ObjectMapper
                    ObjectMapper objectMapper = new ObjectMapper();
                    try {
                        // Convert JSON array string to List of Person objects
                        List<DataValue> dataValues = objectMapper.readValue(metricArray.toJSONString(), objectMapper.getTypeFactory().constructCollectionType(List.class, DataValue.class));
                        if (dataValues != null) {
                            result.put(toCamelCase(key), collectClusterMetrics(key, dataValues));
                        }
                    } catch (JsonProcessingException e) {
                        e.printStackTrace();
                    }
                }



               String localNodeId = NodeUtil.getNodeId(brokerContext);
                ConditionGroup conditionGroup = new ConditionGroup().add(new ConditionQ(MqttMetricCache::getNodeId, localNodeId));

                MqttMetricCache mqttMetricEntity = mqttMetricService.selectOne(conditionGroup, MqttMetricCache.class);
                if (Func.isEmpty(mqttMetricEntity)) {
                    mqttMetricEntity = new MqttMetricCache();
                    Timestamp timestamp = new Timestamp(System.currentTimeMillis());
                    mqttMetricEntity.setUpAt(timestamp);
                }
                mqttMetricEntity.setNodeId(localNodeId);
                //指标信息
                mqttMetricEntity.setMetric(JSON.toJSONString(result));
                //配置信息
                mqttMetricEntity.setConfig(JSONObject.toJSONString(brokerContext.Options()));
                //jvm信息
                mqttMetricEntity.setJvmCpu(JSON.toJSONString(SystemInfoUtil.getSystemInfo()));
                mqttMetricEntity.setUpdatedAt(new Timestamp(System.currentTimeMillis()));

                Map<String, String> infoMap = new HashMap<>();
                // 获取操作系统架构（例如：x86、x64 等），并设置
                infoMap.put("osArch", System.getProperty("os.arch"));
                // 获取操作系统名称（例如：Windows、Linux 等），并设置
                infoMap.put("osName", System.getProperty("os.name"));
                // 获取 Java 虚拟机供应商名称并设置
                infoMap.put("vmVendor", System.getProperty("java.vendor"));
                // 获取 Java 虚拟机版本并设置
                infoMap.put("vmVersion", System.getProperty("java.version"));
                infoMap.put("version", brokerContext.Options().VERSION);
                //其他
                mqttMetricEntity.setOther(JSONObject.toJSONString(infoMap));



                mqttMetricEntity.setStatus(BrokerStatueEnum.RUNNING.getCode());
                // vmVersion
                mqttMetricService.getIgniteCache().put(localNodeId, mqttMetricEntity);

            } catch (Exception e) {

            }
        }, 5L, TimeUnit.SECONDS);

    }


    private static ClientMetrics collectClusterMetrics(String key, List<DataValue> dataValues) {
        ClientMetrics clientOnline = null;
        // 检查 key 是否匹配
        if (key.equals(MqttMetricEnum.BYTES_RECEIVED.getCode())) {
            // 复制并修改 ClientMetrics 的实例，避免修改静态实例
            clientOnline = ClientMetrics.Metrics.BYTES_RECEIVED;

            clientOnline.setX(dataValues.stream().map(dataValue -> String.valueOf(dataValue.getDate())).collect(Collectors.toList()));
            clientOnline.setY(dataValues.stream().map(dataValue -> dataValue.getValue()).collect(Collectors.toList()));
            double latestDataValue = dataValues.get(dataValues.size() - 1).value; // 获取最后一个元素
            clientOnline.setCurrent(SystemInfoUtil.formatMemory((long) latestDataValue)); // 设置 current
        }
        if (key.equals(MqttMetricEnum.BYTES_SENT.getCode())) {
            // 复制并修改 ClientMetrics 的实例，避免修改静态实例
            clientOnline = ClientMetrics.Metrics.BYTES_SENT;

            clientOnline.setX(dataValues.stream().map(dataValue -> String.valueOf(dataValue.getDate())).collect(Collectors.toList()));
            clientOnline.setY(dataValues.stream().map(dataValue -> dataValue.getValue()).collect(Collectors.toList()));
            double latestDataValue = dataValues.get(dataValues.size() - 1).value; // 获取最后一个元素
            clientOnline.setCurrent(SystemInfoUtil.formatMemory((long) latestDataValue)); // 设置 current

        }
        if (key.equals(MqttMetricEnum.CLIENT_ONLINE.getCode())) {
            // 复制并修改 ClientMetrics 的实例，避免修改静态实例
            clientOnline = ClientMetrics.Metrics.CLIENT_ONLINE;

            clientOnline.setX(dataValues.stream().map(dataValue -> String.valueOf(dataValue.getDate())).collect(Collectors.toList()));
            clientOnline.setY(dataValues.stream().map(dataValue -> dataValue.getValue()).collect(Collectors.toList()));
            double latestDataValue = dataValues.get(dataValues.size() - 1).value; // 获取最后一个元素
            clientOnline.setCurrent(String.valueOf((int) latestDataValue)); // 设置 current

        }
        if (key.equals(MqttMetricEnum.PACKETS_PUBLISH_RECEIVED.getCode())) {
            // 复制并修改 ClientMetrics 的实例，避免修改静态实例
            clientOnline = ClientMetrics.Metrics.PACKETS_PUBLISH_RECEIVED;

            clientOnline.setX(dataValues.stream().map(dataValue -> String.valueOf(dataValue.getDate())).collect(Collectors.toList()));
            clientOnline.setY(dataValues.stream().map(dataValue -> dataValue.getValue()).collect(Collectors.toList()));
            double latestDataValue = dataValues.get(dataValues.size() - 1).value; // 获取最后一个元素
            clientOnline.setCurrent(String.valueOf((int) latestDataValue)); // 设置 current

        }
        if (key.equals(MqttMetricEnum.PACKETS_PUBLISH_SENT.getCode())) {
            // 复制并修改 ClientMetrics 的实例，避免修改静态实例
            clientOnline = ClientMetrics.Metrics.PACKETS_PUBLISH_SENT;

            clientOnline.setX(dataValues.stream().map(dataValue -> String.valueOf(dataValue.getDate())).collect(Collectors.toList()));
            clientOnline.setY(dataValues.stream().map(dataValue -> dataValue.getValue()).collect(Collectors.toList()));
            double latestDataValue = dataValues.get(dataValues.size() - 1).value; // 获取最后一个元素
            clientOnline.setCurrent(String.valueOf((int) latestDataValue)); // 设置 current

        }
        if (key.equals(MqttMetricEnum.PACKETS_RECEIVED.getCode())) {
            // 复制并修改 ClientMetrics 的实例，避免修改静态实例
            clientOnline = ClientMetrics.Metrics.PACKETS_RECEIVED;

            clientOnline.setX(dataValues.stream().map(dataValue -> String.valueOf(dataValue.getDate())).collect(Collectors.toList()));
            clientOnline.setY(dataValues.stream().map(dataValue -> dataValue.getValue()).collect(Collectors.toList()));
            double latestDataValue = dataValues.get(dataValues.size() - 1).value; // 获取最后一个元素
            clientOnline.setCurrent(String.valueOf((int) latestDataValue)); // 设置 current

        }

        if (key.equals(MqttMetricEnum.SUBSCRIBE_RELATION.getCode())) {
            // 复制并修改 ClientMetrics 的实例，避免修改静态实例
            clientOnline = ClientMetrics.Metrics.SUBSCRIBE_RELATION;

            clientOnline.setX(dataValues.stream().map(dataValue -> String.valueOf(dataValue.getDate())).collect(Collectors.toList()));
            clientOnline.setY(dataValues.stream().map(dataValue -> dataValue.getValue()).collect(Collectors.toList()));
            double latestDataValue = dataValues.get(dataValues.size() - 1).value; // 获取最后一个元素
            clientOnline.setCurrent(String.valueOf((int) latestDataValue)); // 设置 current

        }
        if (key.equals(MqttMetricEnum.PACKETS_PUBLISH_RATE.getCode())) {
            // 复制并修改 ClientMetrics 的实例，避免修改静态实例
            clientOnline = ClientMetrics.Metrics.PACKETS_PUBLISH_RATE;

            clientOnline.setX(dataValues.stream().map(dataValue -> String.valueOf(dataValue.getDate())).collect(Collectors.toList()));


            List<Double> collect = dataValues.stream().map(dataValue -> dataValue.getValue()).collect(Collectors.toList());
            List<Double> packetsPublishRate = new ArrayList<>();
            collect.forEach(d -> {
                BigDecimal rate = new BigDecimal(d)
                        .divide(new BigDecimal(10), 2, RoundingMode.HALF_UP);  // 除以 100，并保留两位小数
                packetsPublishRate.add(rate.doubleValue());
            });

            clientOnline.setY(packetsPublishRate);
            double latestDataValue = dataValues.get(dataValues.size() - 1).value; // 获取最后一个元素
            // 转换为百分比并保留2位小数
            BigDecimal rate = new BigDecimal(latestDataValue)
                    .divide(new BigDecimal(10), 2, RoundingMode.HALF_UP);  // 除以 100，并保留两位小数
            clientOnline.setCurrent(String.valueOf(rate.doubleValue())); // 设置 current

        }

        if (key.equals(MqttMetricEnum.TOPIC_COUNT.getCode())) {
            // 复制并修改 ClientMetrics 的实例，避免修改静态实例
            clientOnline = ClientMetrics.Metrics.TOPIC_COUNT;

            clientOnline.setX(dataValues.stream().map(dataValue -> String.valueOf(dataValue.getDate())).collect(Collectors.toList()));
            clientOnline.setY(dataValues.stream().map(dataValue -> dataValue.getValue()).collect(Collectors.toList()));
            double latestDataValue = dataValues.get(dataValues.size() - 1).value; // 获取最后一个元素
            clientOnline.setCurrent(String.valueOf((int) latestDataValue)); // 设置 current

        }

        return clientOnline;
    }


    public static String toCamelCase(String str) {
        StringBuilder result = new StringBuilder();
        boolean nextUpperCase = false;

        for (char c : str.toCharArray()) {
            if (c == '_') {
                nextUpperCase = true;
            } else {
                if (nextUpperCase) {
                    result.append(Character.toUpperCase(c));
                    nextUpperCase = false;
                } else {
                    result.append(Character.toLowerCase(c));
                }
            }
        }
        return result.toString();
    }


}
